Upstream:
apache#19924

PR Description

Summary

This PR introduces a high-performance streaming architecture for reading JSON array format files ([{...}, {...}, ...]) in DataFusion. The new design processes arbitrarily large JSON array files with constant memory usage (~32MB) instead of loading the entire file into memory.

Motivation

The previous implementation had critical limitations:

1. Memory explosion: Reading a 33GB JSON array file required 100GB+ memory (full load + serde_json parsing + intermediate strings)
2. Triple parsing overhead: Content was parsed 3 times (serde_json → String → arrow-json)
3. No streaming support: Couldn't process files larger than available memory

Solution Architecture

The new design uses a streaming character substitution approach that converts JSON array format to NDJSON on-the-fly:

┌─────────────────────────────────────────────────────────────────────────┐
│                         S3 / Object Store (async)                       │
│                    (Large JSON Array File, e.g. 33GB)                   │
└─────────────────────────────────────────────────────────────────────────┘
                                │
                                ▼ async stream (Bytes chunks)
┌─────────────────────────────────────────────────────────────────────────┐
│                      Async Task (tokio runtime)                         │
│         Reads chunks from object store, sends to sync channel           │
└─────────────────────────────────────────────────────────────────────────┘
                                │
                                ▼ std::sync::mpsc::sync_channel<Bytes>
                                │   (bounded, ~16MB buffer)
                                ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                   Blocking Task (spawn_blocking)                        │
│  ┌──────────────┐   ┌────────────────────────┐   ┌──────────────────┐  │
│  │ChannelReader │ → │JsonArrayToNdjsonReader │ → │ Arrow JsonReader │  │
│  │   (Read)     │   │  [{},...] → {}\n{}     │   │  (RecordBatch)   │  │
│  └──────────────┘   └────────────────────────┘   └──────────────────┘  │
└─────────────────────────────────────────────────────────────────────────┘
                                │
                                ▼ tokio::sync::mpsc::channel<RecordBatch>
┌─────────────────────────────────────────────────────────────────────────┐
│                      ReceiverStream (async)                             │
│                   → DataFusion execution engine                         │
└─────────────────────────────────────────────────────────────────────────┘

Key Components

1. JsonArrayToNdjsonReader: A streaming Read + BufRead adapter that performs character-level transformation:

   • Skips leading [ and trailing ]
   • Converts top-level , to \n
   • Tracks nesting depth to preserve nested arrays/objects
   • Handles JSON strings correctly (including escaped quotes)

2. ChannelReader: Bridges async-to-sync boundary by receiving Bytes chunks from a channel and implementing std::io::Read

3. JsonArrayStream: Custom stream wrapper that holds SpawnedTask handles for proper cancel-safety

Memory Budget (~32MB total)

API Changes

• Renamed format_array option to newline_delimited (inverted semantics for clarity)
  • newline_delimited: true (default) → NDJSON format
  • newline_delimited: false → JSON array format
• Renamed NdJsonReadOptions to JsonReadOptions (with deprecation alias)
• Removed unused compression_level field from JsonOptions

Testing

• Added comprehensive unit tests for JsonArrayToNdjsonReader including:
  • Nested objects and arrays
  • Escaped quotes in strings
  • Buffer boundary handling
  • Empty arrays
  • Trailing content validation
• Added integration tests for JsonOpener with JSON array format
• Updated existing tests to use new API

Limitations

• JSON array format does not support range-based file scanning (returns clear error)
• Validation of JSON structure is best-effort (malformed JSON may produce partial results)